package net.bwie.realtime.jtp.dws.log.function;

import net.bwie.realtime.jtp.common.utils.DateTimeUtil;
import net.bwie.realtime.jtp.dws.log.been.PageViewBean;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class PageViewWindowFunction implements WindowFunction<PageViewBean,String,String, TimeWindow> {
    @Override
    public void apply(String s, TimeWindow window, Iterable<PageViewBean> input, Collector<String> out) throws Exception {
        String window_start_time = DateTimeUtil.convertLongToString(window.getStart(), DateTimeUtil.DATE_TIME_FORMAT);
        String window_end_time = DateTimeUtil.convertLongToString(window.getEnd(), DateTimeUtil.DATE_TIME_FORMAT);

        PageViewBean pageViewBean = input.iterator().next();
        long svCount = pageViewBean.getSessionCount();
        long pvCount = pageViewBean.getPvCount();
        long pvDuringTime = pageViewBean.getPvDuringTime();
        long uvCount = pageViewBean.getUvCount();

//        // 定义初始值
//        long svCount = 0L;
//        long pvCount = 0L;
//        long pvDuringTime = 0L;
//        long uvCount = 0L;
//        // 迭代聚合数据
//        for (PageViewBean pageViewBean : input) {
//            svCount += pageViewBean.getSessionCount();
//            pvCount += pageViewBean.getPvCount();
//            pvDuringTime += pageViewBean.getPvDuringTime();
//            uvCount += pageViewBean.getUvCount();
//        }

        // 构建输出数据
        String output = window_start_time + "," +
                window_end_time + "," +
                s + "," +
                svCount + "," +
                pvCount + "," +
                pvDuringTime + "," +
                uvCount + "," +
                System.currentTimeMillis();
        out.collect(output);
    }
}
